package dacp.etl.kafka.hdfs.connect.names;

import java.util.Map;

import org.apache.kafka.common.TopicPartition;

import dacp.etl.kafka.hdfs.connect.model.FileNamePropertiesVO;
import dacp.etl.kafka.hdfs.connect.utils.CommonSinkConnectorConfig;
import io.confluent.connect.hdfs.HdfsSinkConnecorConstants;

public class DefaultNames implements FileName{

	private String zeroPadFormat; 
	
	
	@Override
	public void configure(Map<String, Object> config) {
		
		zeroPadFormat = "%0" +
				(int)config.get(CommonSinkConnectorConfig.FILENAME_OFFSET_ZERO_PAD_WIDTH_CONFIG) +
	            "d";
		
	}

	@Override
	public String generateName(TopicPartition topicPart, long startOffset, long endOffset, String directory, String extension) {
		String topic = topicPart.topic();
		int partition = topicPart.partition();
		StringBuilder sb = new StringBuilder();
		sb.append(topic);
		sb.append(HdfsSinkConnecorConstants.COMMMITTED_FILENAME_SEPARATOR);
		sb.append(partition);
		sb.append(HdfsSinkConnecorConstants.COMMMITTED_FILENAME_SEPARATOR);
		sb.append(String.format(zeroPadFormat, startOffset));
		sb.append(HdfsSinkConnecorConstants.COMMMITTED_FILENAME_SEPARATOR);
		sb.append(String.format(zeroPadFormat, endOffset));
		sb.append(extension);
		String name = sb.toString();
		return name;
	}
 

	@Override
	public String generateName(FileNamePropertiesVO vo) {
		return vo.getOnlyId() + vo.getExtension();
	}
 

	

}
